package org.test.rocketmq.consumer02;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * Pull消费:Lite Pull Consumer: subscribe
 */
public class LitePullConsumerSubscribe {
    public static volatile boolean running = true;

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_group_3");
        litePullConsumer.setNamesrvAddr("localhost:9876");
        litePullConsumer.setPullBatchSize(10);
        litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET );
        litePullConsumer.subscribe("BatchTest", "*");
        litePullConsumer.start();
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                System.out.printf("%s%n", messageExts);
                for (MessageExt messageExt : messageExts) {
                    System.out.printf("%s%n", new String(messageExt.getBody(), StandardCharsets.UTF_8));
                }
                Thread.sleep(2000);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
}
